package com.ailk.bigdata.etl.realstream.server.service.impl;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.OutputStreamWriter;
import java.io.UnsupportedEncodingException;
import java.util.Calendar;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.io.FilenameUtils;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.common.server.constant.Constant;
import com.ailk.bigdata.etl.realstream.server.model.RingBufferQueue;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

/**
 * 
 * @description 接口单元写文件线程
 * @author [xuwei3]
 * @version [版本号,2015-7-28]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public class RollingFileTask implements Runnable
{
	private static final Logger logger = LoggerFactory.getLogger(RollingFileTask.class);
	private static final String inUseSuffix = ".tmp";

	private final int rollInterval;
	private final int rollSize;
	private final int batchSize;

	private volatile boolean shouldRotate;

	private BufferedWriter fileWriter;
	private ScheduledExecutorService rollService;

	private AtomicInteger fileExtensionCounter;

	private String lastFileName;// 上一次写入文件文件名

	private volatile String filePath;
	private volatile String fileName;
	private volatile String bucketPath;

	private RingBufferQueue<String> loadingDataQueue = new RingBufferQueue<String>(String.class, 0);

	protected RollingFileTask(int rollInterval, int rollSize, int batchSize, String filePath, String fileName)
	{
		this.rollInterval = rollInterval;
		this.rollSize = rollSize;
		this.batchSize = batchSize;
		this.filePath = filePath;
		this.fileName = fileName;
		fileExtensionCounter = new AtomicInteger(0);
		shouldRotate = false;
	}

	/**
	 * @description 往落地处理线程放入消息
	 * @param realData
	 */
	public void add(String realData)
	{
		this.loadingDataQueue.enQueue(realData);
	}

	/**
	 * @description 重命名文件，去除临时文件标示
	 * @param bucketPath
	 */
	private void rename(String bucketPath)
	{
		File writedFile = new File(bucketPath);
		writedFile.renameTo(new File(StringUtils.removeEnd(bucketPath, inUseSuffix)));
	}

	/**
	 * @description 获取当前文件名称
	 */
	private void getCurrentFileName()
	{
		int counter = 0;
		String fullFilePath = FilePathFormatUtil.escapeString(filePath, true, Calendar.HOUR_OF_DAY, 1);
		String fullFileName = FilePathFormatUtil.escapeString(FilenameUtils.getBaseName(fileName), true, Calendar.MINUTE, rollInterval)
				+ FilenameUtils.EXTENSION_SEPARATOR_STR + FilenameUtils.getExtension(fileName);

		if (StringUtils.isNotEmpty(lastFileName) && lastFileName.equalsIgnoreCase(fullFileName))
		{
			counter = fileExtensionCounter.incrementAndGet();
		}
		else
		{
			fileExtensionCounter.set(0);
		}

		lastFileName = fullFileName;

		bucketPath = fullFilePath + File.separator + String.format(fullFileName, counter) + inUseSuffix;
		logger.info("创建接口落地文件路径[" + bucketPath + "]......");
	}

	public int procFileWriter() throws UnsupportedEncodingException, FileNotFoundException
	{
		if (shouldRotate)
		{
			if (fileWriter != null)
			{
				IOUtils.closeQuietly(fileWriter);
				fileWriter = null;
				shouldRotate = false;
				rename(bucketPath);
			}
		}
		if (fileWriter == null)
		{
			getCurrentFileName();
			File currentFile = new File(bucketPath);
			fileWriter = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(currentFile), "UTF-8"), 65536);
		}

		int status = Constant.RETURN_OK;

		try
		{
			String tmpData = null;
			int count = 0;
			for (;;)
			{
				tmpData = loadingDataQueue.deQueue();

				fileWriter.write(tmpData);
				fileWriter.newLine();

				if (count++ >= batchSize || shouldRotate)
				{
					fileWriter.flush();
					count = 0;
					if (shouldRotate)
					{
						break;
					}
				}

				tmpData = null;
			}
		}
		catch (Exception e)
		{
			status = Constant.RETURN_FAILD;
			logger.warn("写入落地文件异常", e);
		}
		return status;
	}

	public void initRollIntervalTh()
	{
		// 如果按照时间间隔分批文件
		if (rollInterval > 0)
		{
			logger.info("处理文件定时分批线程启动");
			rollService = Executors.newScheduledThreadPool(1,
					new ThreadFactoryBuilder().setNameFormat("rollingFileSink-roller-" + Thread.currentThread().getId() + "-%d").build());

			rollService.scheduleAtFixedRate(new Runnable()
			{

				@Override
				public void run()
				{
					shouldRotate = true;
				}

			}, rollInterval, rollInterval, TimeUnit.MINUTES);
		}
	}

	public void stop()
	{
		IOUtils.closeQuietly(fileWriter);

		if (rollInterval > 0)
		{
			rollService.shutdown();

			while (!rollService.isTerminated())
			{
				try
				{
					rollService.awaitTermination(1, TimeUnit.SECONDS);
				}
				catch (InterruptedException e)
				{
					logger.debug("Interrupted while waiting for roll service to stop. " + "Please report this.", e);
				}
			}
		}
	}

	@Override
	public void run()
	{
		logger.info("落地文件处理线程启动任务[" + Thread.currentThread().getId() + "],批量[" + batchSize + "]");
		initRollIntervalTh();
		while (true)
		{
			try
			{
				if (Constant.RETURN_FAILD == procFileWriter())
				{
					Thread.sleep(10);
				}
			}
			catch (Exception e)
			{
				logger.error("", e);
			}
		}
	}

}
